package com.ost.kautilya.modeling.weka;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import weka.classifiers.Classifier;
import weka.classifiers.evaluation.NumericPrediction;
import weka.classifiers.timeseries.WekaForecaster;
import weka.classifiers.timeseries.core.TSLagMaker;
import weka.classifiers.timeseries.eval.TSEvaluation;
import weka.core.Attribute;
import weka.core.DenseInstance;
import weka.core.Instance;
import weka.core.Instances;

import com.ost.kautilya.modeling.AttributeInfo;
import com.ost.kautilya.modeling.DataRow;
import com.ost.kautilya.modeling.Forecaster;
import com.ost.kautilya.modeling.ForecasterPeriodicity;
import com.ost.kautilya.modeling.ModelBinary;
import com.ost.kautilya.modeling.ModelHandler;
import com.ost.kautilya.modeling.ModelParameters;
import com.ost.kautilya.modeling.Periodicity;
import com.ost.kautilya.utils.Metric;
import com.ost.kautilya.utils.Metric.DataType;
import com.ost.kautilya.utils.StatisticalMetric;

public class WekaTSForcastingModel extends WekaClassifierModel implements Forecaster {
	private final static Logger logger = Logger.getLogger(WekaTSForcastingModel.class.getName());
	private static final long serialVersionUID = 1L;
	private static final String FORECASTER_BINARY = "forecaster";

	private transient int availablePredictionSize = 0;

	protected transient Instances fins = null;

	private transient long lastOpTime = Long.MIN_VALUE;
	private transient StatisticalMetric<Number>[] dummyDataProvider = null;

	private transient boolean canForecast = true;
	private boolean canPrime = false;
	private Map<Integer, StatisticalMetric<Number>> aDummy = new Hashtable<Integer, StatisticalMetric<Number>>();
	private boolean forcePredictions;
	private transient Instances odis = null;

	public WekaTSForcastingModel(Class<?> cfc) throws Exception {
		super(cfc, ModelType.FORECASTING);
	}

	@Override
	protected void finalize() throws Throwable {
	}

	@Override
	public ModelBinary generateBinary() throws Exception {
		ModelBinary bin = super.generateBinary();
		bin.put(FORECASTER_BINARY, new WekaForecaster());
		return bin;
	}

	@Override
	protected Classifier getClassifier() {
		return getForecasterBinary().getBaseForecaster();
	}

	@Override
	public void finalizeTraining(DataRow tr) throws Exception {
		try {
			WekaForecaster ftr = getForecasterBinary();
			Instances ins = getTrainingInstances();
			if (ins.size() <= 1)
				return;
			for (int i = 0; i < getClasses().length; i++)
				ins.setClassIndex(i);
			ArrayList<Attribute> wAtt = new ArrayList<Attribute>();
			for (int i = 0; i < ins.numAttributes(); i++) {
				Attribute att = ins.attribute(i);
				wAtt.add(att);
			}
			ModelHandler rpt = getHandler();
			List<String> flds2lag = new ArrayList<String>();
			List<String> offsetflds = new ArrayList<String>();
			String clss = "";
			int tsf = -1;
			for (int i = 0; i < ins.numAttributes(); i++) {
				Attribute att = ins.attribute(i);
				String colName = att.name();
				boolean isClass = false;
				for (int j = 0; j < getClasses().length; j++) {
					if (colName.equalsIgnoreCase(rpt.getAttributeInfoByUID(getClasses()[j]).getAttributeTitle())) {
						clss += (clss.length() == 0 ? "" : ",") + colName;
						flds2lag.add(colName);
						isClass = true;
						break;
					}
				}
				if (isClass)
					continue;
				else if (att.type() != Attribute.DATE)
					offsetflds.add(colName);
				else
					tsf = i;
			}
			setParameter(ModelParameters.TSF_TIME_ATTRIBUTE_INDEX, String.valueOf(tsf - getClasses().length));
			int np = Integer.parseInt(getParameter(ModelParameters.TSF_NUMBER_OF_PREDICTIONS));
			ftr.setBaseForecaster(getClassifier());
			WekaForecaster forecaster = ftr;
			{ // simple configuration
				if (forecaster != null) {
					TSLagMaker lagMaker = forecaster.getTSLagMaker();

					String selected = ins.attribute(tsf).name();
					lagMaker.setTimeStampField(selected);
					lagMaker.setAdjustForTrends(true);

					lagMaker.setAddAMIndicator(false);
					lagMaker.setAddDayOfWeek(false);
					lagMaker.setAddMonthOfYear(false);
					lagMaker.setAddQuarterOfYear(false);
					lagMaker.setAddWeekendIndicator(false);

					if (!"AUTO".equalsIgnoreCase(getParameter(ModelParameters.TSF_PERIODICITY))) {
						ftr.getTSLagMaker().setPeriodicity(weka.classifiers.timeseries.core.TSLagMaker.Periodicity.valueOf(getParameter(ModelParameters.TSF_PERIODICITY)));
						forecaster.getTSLagMaker().setSkipEntries("");
					}
					if (getParameter(ModelParameters.TSF_CALCULATE_CONFIDENCE_INTERVALS) != null) {
						forecaster.setCalculateConfIntervalsForForecasts(np);
						double confLevel = Double.valueOf(getParameter(ModelParameters.TSF_CONFIDENCE_INTERVAL_LEVEL));
						forecaster.setConfidenceLevel(confLevel / 100.0);
					} else {
						forecaster.setCalculateConfIntervalsForForecasts(0);
					}
					ftr.setFieldsToForecast(clss);
				}

			}
			{// advanced config
				TSLagMaker lagMaker = forecaster.getTSLagMaker();

				// Set variance adjustment
				lagMaker.setAdjustForVariance(getParameter(ModelParameters.TSF_ADJUST_FOR_VARIANCE) != null);

				int minLag = Integer.parseInt(getParameter(ModelParameters.TSF_MIN_LAG));
				int maxLag = Math.min(Integer.parseInt(getParameter(ModelParameters.TSF_MAX_LAG)), ins.size() / 2);
				{
					// set lag lengths from spinners
					if (maxLag < minLag) {
						throw new Exception("Maximum lag value must be greater than or equal to" + " the minimum lag value");
					}
					Instances m_instances = ins;
					if (m_instances != null && maxLag > m_instances.numInstances()) {
						throw new Exception("The maximum lag can't exceed the number instances (" + m_instances.numInstances() + ") in the data!");
					}

					lagMaker.setMinLag(minLag);
					lagMaker.setMaxLag(maxLag);
					String m_fineTuneLagsField = getParameter(ModelParameters.TSF_FINE_TUNE_LAGS);
					if (m_fineTuneLagsField != null && m_fineTuneLagsField.length() > 0) {
						lagMaker.setLagRange(m_fineTuneLagsField);
					} else {
						lagMaker.setLagRange("");
					}

					// set up averaging from spinners
					if (getParameter(ModelParameters.TSF_AVERAGE_CONSECUTIVE_LONG_LAGS) != null) {
						lagMaker.setAverageConsecutiveLongLags(true);

						int avLagsAfter = Integer.parseInt(getParameter(ModelParameters.TSF_AVERAGE_LAGS_AFTER));
						int numToAv = Integer.parseInt(getParameter(ModelParameters.TSF_NUM_CONSECUTIVE_LONG_LAGS));

						if (avLagsAfter < minLag || avLagsAfter > maxLag) {
							throw new Exception("Point at which to start lag averaging must " + "lie between the minimum and maximum lag value.");
						}

						lagMaker.setAverageLagsAfter(avLagsAfter);
						lagMaker.setNumConsecutiveLongLagsToAverage(numToAv);
					} else {
						lagMaker.setAverageConsecutiveLongLags(false);
					}
				}

				// date-derived periodic customization
				// reset all to false
				lagMaker.setAddAMIndicator(getParameter(ModelParameters.TSF_AM_INDICATOR) != null);
				lagMaker.setAddDayOfWeek(getParameter(ModelParameters.TSF_ADD_DAY_OF_WEEK) != null);
				lagMaker.setAddMonthOfYear(getParameter(ModelParameters.TSF_ADD_MONTH_OF_YEAR) != null);
				lagMaker.setAddQuarterOfYear(getParameter(ModelParameters.TSF_ADD_QUARTER_OF_YEAR) != null);
				lagMaker.setAddWeekendIndicator(getParameter(ModelParameters.TSF_ADD_WEEKEND_INDICATOR) != null);
				lagMaker.setAddDayOfMonth(getParameter(ModelParameters.TSF_ADD_DAY_OF_MONTH) != null);
				lagMaker.setAddNumDaysInMonth(getParameter(ModelParameters.TSF_ADD_NUM_DAYS_IN_MONTH) != null);
				if (lagMaker.getAddAMIndicator() || lagMaker.getAddDayOfWeek() || lagMaker.getAddMonthOfYear() || lagMaker.getAddQuarterOfYear() || lagMaker.getAddWeekendIndicator() || lagMaker.getAddDayOfMonth() || lagMaker.getAddNumDaysInMonth()) {
					lagMaker.clearCustomPeriodics();
				}

				// non-date primary periodic attribute
				/*String ppfn = m_primaryPeriodicCombo.getSelectedItem().toString();
				if (!ppfn.equals("<None>")) {
					lagMaker.setPrimaryPeriodicFieldName(ppfn);
				} else {
				}*/
				lagMaker.setPrimaryPeriodicFieldName("");


				if (offsetflds.size() > 0) {
					String overlays = "";
					for (int i = 0; i < offsetflds.size(); i++) {
						String s = offsetflds.get(i);
						overlays += (i == 0 ? "" : ",") + s;
					}
					ftr.setOverlayFields(overlays);
				} else {
					ftr.setOverlayFields(null);
				}
			}

			System.out.println("building forecaster " + getBinaryName() + " with " + ins.size() + " row(s), relation - " + ins.relationName());
			// System.out.println(ins);

			ftr.buildForecaster(ins, System.out);
		} finally {
		}
	}

	private WekaForecaster getForecasterBinary() {
		return (WekaForecaster) getBinary().get(FORECASTER_BINARY);
	}

	@Override
	public void finalizeEvaluation(DataRow tr) throws Exception {
		ByteArrayOutputStream out = new ByteArrayOutputStream();
		try {
			if (getEvaluationInstances().size() == 0) {
				return;
			}
			// for poa adjustment

			Instances tins = getTrainingInstances();
			Instances vins = getEvaluationInstances();
			TSEvaluation ev = new TSEvaluation(tins, vins);
			try {
				ev.evaluateForecaster(getForecasterBinary(), new PrintStream(out));
			} catch (Exception e) {
				logger.log(Level.SEVERE, e.getMessage());
			} finally {
				try {
					new PrintStream(out).write(ev.toSummaryString().getBytes());
				} catch (Exception e) {
					logger.log(Level.SEVERE, e.getMessage());
				}
			}
			addEvaluation(tr, new WekaTSFEvaluation(ev, Integer.parseInt(getParameter(ModelParameters.TSF_NUMBER_OF_PREDICTIONS)), getClasses().length, out.toByteArray()));
		} finally {
			if (out != null) {
				out.close();
			}
		}
	}

	@Override
	public void prepare(ModelHandler rpt, ModelOperationMode mode) throws Exception {
		// System.out.println("preparing forecaster to " + getMode());
		super.prepare(rpt, mode);
		forcePredictions = false;
		setParameter(ModelParameters.TSF_T_LAST, String.valueOf(Long.MAX_VALUE));
		if (mode != ModelOperationMode.PREDICT) {
			setParameter(ModelParameters.TSF_DATA_PERIOD_UNITS, null);
		}

		this.fins = new Instances(getModelId(), getWekaAttributes(), 0);
		for (int cli = 0; cli < getClasses().length; cli++) {
			this.fins.setClassIndex(cli);
		}
		availablePredictionSize = Integer.parseInt(getParameter(ModelParameters.TSF_NUMBER_OF_PREDICTIONS));
		boolean primeWithDummyAverage = getParameter(ModelParameters.TSF_CONCIDER_MISSING_ROWS_AS_AVERAGE_TILL_POINT) != null;
		for (int i = 0; i < getAttributes().length; i++) {
			AttributeInfo fld = rpt.getAttributeInfoByUID(getAttributes()[i]);
			if (fld.getAttributeDataType().equalsIgnoreCase("date")) {
				setParameter(ModelParameters.TSF_TIME_ATTRIBUTE_INDEX, String.valueOf(i));
			} else if (!fld.getAttributeDataType().equalsIgnoreCase("text")) {
				if (primeWithDummyAverage)
					aDummy.put(i, (StatisticalMetric<Number>) Metric.newMetric(Metric.MEAN, DataType.DOUBLE));
			}
		}
		if (primeWithDummyAverage) {
			dummyDataProvider = new StatisticalMetric[getClasses().length];
			for (int i = 0; i < dummyDataProvider.length; i++)
				dummyDataProvider[i] = (StatisticalMetric<Number>) Metric.newMetric(Metric.MEAN, DataType.DOUBLE);
		}
		lastOpTime = Long.MIN_VALUE;
		if (mode == ModelOperationMode.PREDICT) {
			setCanPredict(true);
			setCanPrime(true);
			try {
				getForecasterBinary().primeForecaster(fins);
			} catch (Exception e) {
				logger.severe(e.getMessage());
			}
		}

		if (hasOverlays()) {
			this.odis = new Instances(getModelId(), getWekaAttributes(), 0);
			for (int cli = 0; cli < getClasses().length; cli++) {
				this.odis.setClassIndex(cli);
			}
		}
	}

	@Override
	public Instances getTrainingInstances() {
		return super.getTrainingInstances();
	}

	@Override
	public Object[] predictClasses(DataRow rs) throws Exception {
		Object[][] value = new Object[getClasses().length][availablePredictionSize];
		List<List<NumericPrediction>> list = generatePredictions(rs);
		for (int cli = 0; cli < getClasses().length; cli++) {
			for (int i = 0; i < availablePredictionSize; i++) {
				Number val = 0d;
				try {
					val = (Number) (list.get(i).get(cli)).predicted();
				} catch (Exception e) {
					logger.log(Level.WARNING, e.getMessage());
					val = 0;
				} finally {
					value[cli][i] = val.doubleValue();
				}
			}
		}
		try {
			return value;
		} finally {
		}
	}

	private List<List<NumericPrediction>> generatePredictions(DataRow rs) {
		List<List<NumericPrediction>> predictions = new ArrayList<List<NumericPrediction>>();
		try {
			WekaForecaster ftr = getForecasterBinary();
			if (!hasOverlays()) {
				primeForecasterIncrementally(rs);
				predictions = ftr.forecast(availablePredictionSize, System.out);
			} else {
				Instances odis = getOverlayData();
				predictions = ftr.forecast(odis.size() < availablePredictionSize ? odis.size() : availablePredictionSize, odis, System.out);
			}
		} catch (Exception e) {
			logger.log(Level.SEVERE, e.getMessage());
		} finally {
		}
		return predictions;
	}

	@Override
	public boolean canProcessRow(DataRow sqlRs) throws Exception {
		ModelOperationMode mode = getMode();
		boolean cp = mode == ModelOperationMode.TRAIN || mode == ModelOperationMode.EVALUATION;
		try {
			if (mode == ModelOperationMode.PREDICT) {
				return forcePredictions;
			}
			return cp;
		} finally {
		}
	}

	@Override
	public void processNotAccountedRow(DataRow tr) throws Exception {
		try {
			ModelOperationMode mode = getMode();
			if (mode == ModelOperationMode.PREDICT) {
				if (canPrime && canForecast) {
					primeForecasterIncrementally(tr);
				} else {
					Instance lastIn = new DenseInstance(getWekaAttributes().size());
					lastIn = updateInstance(tr, lastIn);
					Attribute att = getTimeAttribute();
					setParameter(ModelParameters.TSF_T_LAST, String.valueOf(lastIn.value(att)));
				}
			}
		} finally {
		}
	}

	public String getBinaryName(String[] groups) {
		String name = super.getBinaryName();
		if (splitMode) {
			for (int i = 0; i < groups.length; i++)
				name += "-" + groups[i];
		}
		return name;
	}

	@Override
	public boolean hasOverlays() {
		return getAttributes().length > 1;
	}

	@Override
	public void forceLevelBreak() {
		forcePredictions = true;
	}

	@Override
	public ForecasterPeriodicity getPeriodicity() {
		WekaForecaster ftr = getForecasterBinary();
		return Enum.valueOf(Periodicity.class, ftr == null || ftr.getTSLagMaker() == null || getMode() == ModelOperationMode.TRAIN
				|| ftr.getTSLagMaker().getPeriodicity() == weka.classifiers.timeseries.core.TSLagMaker.Periodicity.UNKNOWN ? getParameter(ModelParameters.TSF_PERIODICITY) : ftr.getTSLagMaker()
				.getPeriodicity().name());
	}

	@Override
	public void initTransients() {
		super.initTransients();
		lastOpTime = Long.MIN_VALUE;
		setCanPredict(false);
		setCanPrime(false);
	}

	@Override
	public String toString() {
		return super.toString() + " - " + getBinaryName();
	}

	@Override
	public void primeForecasterIncrementally(DataRow tr) throws Exception {
		processWithMissingAverages(tr, Operation.PRIME);
	}

	private void updateDummyTotals(DataRow tr) {
		if (dummyDataProvider != null && !tr.isInjectedRow()) {
			for (int cli = 0; cli < getClasses().length; cli++) {
				AttributeInfo atf = getHandler().getAttributeInfoByUID(getClasses()[cli]);
				dummyDataProvider[cli].push(0, ((Number) tr.getObjectArray()[atf.getDataRowIndex()]).doubleValue());
			}
		}
		for (int ati : aDummy.keySet()) {
			AttributeInfo atf = getHandler().getAttributeInfoByUID(getAttributes()[ati]);
			aDummy.get(ati).push(0, ((Number) tr.getObjectArray()[atf.getDataRowIndex()]).doubleValue());
		}
	}

	private Attribute getTimeAttribute() {
		return getWekaAttributes().get(getClasses().length + Integer.parseInt(getParameter(ModelParameters.TSF_TIME_ATTRIBUTE_INDEX)));
	}

	@Override
	public int getTimeAttributeID() throws Exception {
		return getAttributes()[Integer.parseInt(getParameter(ModelParameters.TSF_TIME_ATTRIBUTE_INDEX))];
	}

	@Override
	public int getTimeFieldIndex() {
		return getHandler().getAttributeInfoByUID(getAttributes()[Integer.parseInt(getParameter(ModelParameters.TSF_TIME_ATTRIBUTE_INDEX))]).getDataRowIndex();
	}

	public void setCanPredict(boolean canForecast) {
		this.canForecast = canForecast;
	}

	public static enum Operation {
		TRAIN, EVAL, PRIME, OVERLAY_SET
	}

	@Override
	public void addOverlayDataForPrediction(DataRow tr) throws Exception {
		processWithMissingAverages(tr, Operation.OVERLAY_SET);
	}

	public void processWithMissingAverages(DataRow tr, Operation op) throws Exception {
		Instances lins = new Instances(getModelId(), getWekaAttributes(), 1);
		for (int cli = 0; cli < getClasses().length; cli++) {
			lins.setClassIndex(cli);
		}
		Instance lastIn = new DenseInstance(getWekaAttributes().size());
		lastIn.setDataset(lins);
		lastIn = updateInstance(tr, lastIn);

		Attribute att = getTimeAttribute();
		setParameter(ModelParameters.TSF_T_LAST, String.valueOf(lastIn.value(att)));

		long currentTime = (long) lastIn.value(getTimeAttribute());
		if (currentTime <= lastOpTime) {
			logger.info(getModelId() + " - Skipping already processed row - " + tr + ", for, Operation - " + op);
			return;
		}
		try {
			long diff = currentTime - lastOpTime;
			if (dummyDataProvider != null && lastOpTime != Long.MIN_VALUE) {
				long pct = getPeriodicity().valueInMillis();
				if (diff > pct) {
					double dtdiff = lastOpTime + pct;
					do {
						try {
							Instance dummy = new DenseInstance(getWekaAttributes().size());
							dummy.setDataset(lins);
							dummy.setValue(getTimeAttribute(), dtdiff);
							updateWithAverages(dummy);
							Operation opt = op;
							if (opt == Operation.OVERLAY_SET && getOverlayData().size() == 0) {
								opt = Operation.PRIME;
							}
							performOperation(dummy, opt);
						} catch (Exception e) {
							logger.severe(e.getMessage());
							continue;
						} finally {
							dtdiff += pct;
						}
					} while ((diff -= pct) > pct);
				}
			}
		} finally {
			try {
				performOperation(lastIn, op);
			} finally {
				lastOpTime = currentTime;
				updateDummyTotals(tr);
			}
		}
	}

	@Override
	public void updateLastBinaryName(Object[] groupVals) {
		super.updateLastBinaryName(groupVals);
	}

	@Override
	public long predictedAt() {
		long correction = 0;
		if (hasOverlays()) {
			correction = (getOverlayData().size() - 1) * (getPeriodicity().valueInMillis()) * -1;
		} else {
			correction = getPeriodicity().valueInMillis();
		}
		return lastOpTime + correction;
	}

	private void updateWithAverages(Instance dummy) {
		if (dummyDataProvider != null) {
			for (int cli = 0; cli < getClasses().length; cli++) {
				dummy.setValue(cli, dummyDataProvider[cli].value().doubleValue());
			}
		}
		for (int ati : aDummy.keySet()) {
			dummy.setValue(getClasses().length + ati, aDummy.get(ati).value().doubleValue());
		}
	}

	private void performOperation(Instance lins, Operation op) throws Exception {
		try {
			switch (op) {
			case PRIME:
				getForecasterBinary().primeForecasterIncremental(lins);
				break;
			case TRAIN:
				getTrainingInstances().add(lins);
				break;
			case EVAL:
				getEvaluationInstances().add(lins);
				break;
			case OVERLAY_SET:
				Instances ins = getOverlayData();
				ins.add(lins);
				break;
			}
		} finally {
			// System.out.println(getModelId() + " [" + op + "] @ " + lins);
		}
	}

	private Instances getOverlayData() {
		return odis;
	}

	@Override
	public void evaluateRecordIncrementally(DataRow tr) throws Exception {
		processWithMissingAverages(tr, Operation.EVAL);
	}

	@Override
	public void trainWithRecordIncrementally(DataRow tr) throws Exception {
		processWithMissingAverages(tr, Operation.TRAIN);
	}

	@Override
	public void finalizeBatchPredictions() throws Exception {
		loadBinary(getModelId());
		getBinary().put(RUNTIME_MODE, ModelOperationMode.SILENT);
	}

	@Override
	public ModelOperationMode getMode() {
		ModelOperationMode mode = null;
		return mode == null || getHandler().inBatchMode() ? (mode == ModelOperationMode.EVALUATION ? mode : super.getMode()) : mode;
	}

	@Override
	public void setCanPrime(boolean b) {
		canPrime = b;
	}

	@Override
	public boolean getCanPrime() {
		return canPrime;
	}

	public boolean isBreakingAtLevel() {
		return super.isBreakingAtLevel();
	}

	@Override
	public int getNumericLabel() {
		return -Math.abs(getMetaData().getNumericId());
	}
}
